/*
 * scheduler.cpp
 *
 *  Created on: Aug 10, 2010
 *      Author: sdaniele
 */

#include "scheduler.h"
#include "process.h"
#include "dbgprint.hpp"

#include "boost/concept/assert.hpp"
#include "boost/concept_check.hpp"
#include <boost/type_traits.hpp>
#include <boost/static_assert.hpp>
#include <boost/mpl/assert.hpp>
#include <boost/foreach.hpp>
#define foreach         BOOST_FOREACH
#define reverse_foreach BOOST_REVERSE_FOREACH

#include "tbb/tick_count.h"
#include "tbb/tbb_thread.h"
#include "tbb/task_scheduler_init.h"
#include "tbb/parallel_for.h"
#include "tbb/blocked_range.h"
#include "tbb/parallel_do.h"


#include <stdlib.h>
#include <string.h>
#include <assert.h>


namespace pilot {


template <class RandomAccessContainer>
struct parallel_for_helper{

        void operator()(const tbb::blocked_range<size_t>& r) const {
                for(size_t i=r.begin();i!=r.end();++i)
                	m_elems[i]->go();
        }

        RandomAccessContainer& m_elems;
        parallel_for_helper(RandomAccessContainer& elems):m_elems(elems){}
};





scheduler::scheduler()
	: m_delta(0)
	, m_phase(before_start)
	, m_is_parcontext(false)
{
}

scheduler::~scheduler(){
	assert(m_phase==stopped);
}


void scheduler::stop(){
	assert(m_is_parcontext);
	m_should_stop = true;
}

void scheduler::submit(process *p){
	assert(!m_is_parcontext);
	m_process_queue.push_frontside(p);
}


void scheduler::spawn(process *p){
	m_process_queue.push_backside(p);
}

void scheduler::spawn_group_internal(const event::container_type& group){
	BOOST_CONCEPT_ASSERT((boost::Container< event::container_type >));
	BOOST_STATIC_ASSERT((boost::is_same<event::container_type::value_type,process*>::value));

	foreach(event::container_type::value_type p, group){
		m_process_queue.push_backside(p);
	}
}


void scheduler::start(){
	assert(!m_is_parcontext);
	assert(m_phase==before_start);
	assert(m_delta==0);

	m_phase = running;
	m_started_at = tbb::tick_count::now();

	while(should_continue()){
		cycle();
		m_process_queue.swap();

		for(event_registry_t::iterator i = m_event_registry.begin(); i!= m_event_registry.end(); ++i)
			if((*i)->test_and_reset())
				spawn_group_internal((*i)->get_subscribers());

		++m_delta;
	}
	m_elapsed = (tbb::tick_count::now() - m_started_at);
	m_phase = stopped;
}


void scheduler::cycle(){
	assert(!m_is_parcontext);
	assert(m_phase==running);

	m_is_parcontext = true;
	tbb::parallel_for(
			tbb::blocked_range<size_t>(0,m_process_queue.size())
		  , parallel_for_helper<queue_t>(m_process_queue)
		);
	m_is_parcontext = false;
	dbgprint::dump(m_delta);
}

bool scheduler::should_continue() const {
	assert(!m_is_parcontext);
	assert(m_phase==running);

	return !(m_process_queue.is_exhausted() || m_should_stop);
}

double scheduler::seconds_elapsed() const {
	if(m_phase==before_start)
		return 0;
	if(m_phase==running)
		return (::tbb::tick_count::now() - m_started_at).seconds();
	//if(m_phase==stopped)
	return m_elapsed.seconds();
}

void scheduler::register_event(event* e){
	m_event_registry.push_back(e);
}


}
